﻿
#define STREAM_TRACE

using System;
using System.Collections.Generic;
using System.Runtime.Serialization.Json;
using System.ComponentModel;
using System.IO;
using System.Threading.Tasks;
using System.Linq;
using System.Xml.Linq;
using System.Text;
using Net.Chirash.Twitter.TwitterDatabase;

namespace Net.Chirash.Twitter.TwitterClient
{
    public class TwitterStreamMonitor
        : IObservable<TwitterMessage>, IDisposable
    {
        private Stream listeningStream;
        private TwitterConnection conn;
        private Uri uri;
        private Task monitoringTask;
        private bool monitoringCancellation = false;

        public TwitterStreamMonitor(Stream listening, TwitterConnection conn, Uri uri)
        {
            this.listeningStream = listening;
            this.conn = conn;
            this.uri = uri;
        }

        public void StartMonitoring()
        {
            if (this.monitoringTask != null)
                throw new InvalidOperationException("既にモニタリングスレッドは開始されています。");

            this.monitoringTask = Task.Factory.StartNew(new Action(() => monitoringMethod(defaultInterval)));
        }

        private int defaultInterval
        {
            get { return new Random().Next(20, 40); }
        }

        private void monitoringMethod(int reconnectingWaitSec)
        {
            if (this.listeningStream != null)
            {

                try
                {
#if STREAM_TRACE
                    using (var fs = new FileStream("stream_trace.txt", FileMode.OpenOrCreate))
                    using (var fwr = new StreamWriter(fs))
#endif
                    using (var sr = new StreamReader(this.listeningStream))
                    {
                        reconnectingWaitSec = defaultInterval;
                        while (!sr.EndOfStream && !monitoringCancellation)
                        {
                            string line;
                            try
                            {
                                line = sr.ReadLine();
                                if (line == "") // keep-alive
                                    continue;
                                else
                                    parseReceivedString(line);
#if STREAM_TRACE
                                fwr.WriteLine(line);
#endif
                            }
                            catch (System.Xml.XmlException) { }
                        }
                    }
                }
                catch (System.Net.WebException) { }
                catch (System.Net.Sockets.SocketException) { }
                catch (System.IO.IOException) { }
                catch (Exception ex)
                {
                    Console.WriteLine("Occured at UserStreamObserver");
                    Console.WriteLine(ex.ToString());
                }

                Console.WriteLine("try reconnecting");
                this.listeningStream.Close();
            }
            //reconnecting
            this.listeningStream = null;
            System.Threading.Thread.Sleep(reconnectingWaitSec * 1000);
            try
            {
                this.listeningStream = this.conn.TwitterStreamRequest(this.uri.ToString());
            }
            catch (Exception) { }
            monitoringMethod((reconnectingWaitSec * 2 <= 300) ? reconnectingWaitSec * 2 : (new Random()).Next(240, 300));

        }

        public bool IsMonitoringStarted
        {
            get { return this.monitoringTask != null; }
        }

        public bool IsAvailable
        {
            get { return this.listeningStream != null; }
        }

        private void parseReceivedString(string jsonformatted)
        {
            using (var xmlreader
                = JsonReaderWriterFactory.CreateJsonReader(Encoding.Unicode.GetBytes(jsonformatted), System.Xml.XmlDictionaryReaderQuotas.Max))
            {
                // TODO : string.Contains　で代用可能か検討する
                //TODO : directmessageでもdelete飛んでくるはず
                xmlreader.Read();
                var doc = new System.Xml.XmlDocument();
                doc.LoadXml(xmlreader.ReadOuterXml());

                var contextString = doc.OuterXml;
                var element = XElement.Parse(contextString);
                var message = new TwitterMessage();
                var cmd = new TwitterCommand(conn);

                //translate received message
                if (element.Element("text") != null) // is Status
                {
                    if (element.Element("retweeted_status") != null) //is RetweetedStatus
                    {
                        message.MessageType = TwitterMessageType.RetweetedStatus;
                        message.Data = new Status().Parse(contextString, cmd);
                    }
                    else // is not ReTweetedStatus
                    {
                        message.MessageType = TwitterMessageType.Status;
                        message.Data = new Status().Parse(contextString, cmd);
                    }
                }
                else if (element.Element("direct_message") != null) // is DirectMessage
                {
                    message.MessageType = TwitterMessageType.DirectMessage;
                    message.Data = new DirectMessage().Parse(contextString);
                }
                else if (element.Element("delete") != null) // is DeleteStatus
                {
                    message.MessageType = TwitterMessageType.Status;
                    message.Data = DeletedData.Parse(contextString);
                }
                else if (element.Element("event") != null) // is Event
                {
                    message.MessageType = TwitterMessageType.Event;
                    message.Data = TwitterEvent.Parse(contextString,cmd);
                }
                else if (element.Element("friends") != null) // is CurrentFriends
                {
                    //message.MessageType = TwitterMessageType.CurrentFriends;

                }

                System.Diagnostics.Debug.Assert(message.Data == null);

                //Console.WriteLine(message.MessageType.ToString());
                foreach (var observer in this.observers)
                {
                    observer.OnNext(message);

                }
            }


        }

        #region implemention of IObservable
        private List<IObserver<TwitterMessage>> observers = new List<IObserver<TwitterMessage>>();
        public IDisposable Subscribe(IObserver<TwitterMessage> observer)
        {
            observers.Add(observer);

            return new UnSubscriber<TwitterMessage>(observers, observer);
        }

        #region UnSubscriber class
        private class UnSubscriber<TypeData>
            : IDisposable
        {
            private IList<IObserver<TypeData>> observerList;
            private IObserver<TypeData> observer;

            public UnSubscriber(IList<IObserver<TypeData>> list, IObserver<TypeData> observer)
            {
                this.observerList = list;
                this.observer = observer;
            }

            public void Dispose()
            {
                if (observer != null && observerList.Contains(observer))
                    observerList.Remove(observer);
            }
        }
        #endregion
        #endregion

        #region Dispose-Finalize Pattern

        private bool __disposed = false;
        public void Dispose()
        {
            GC.SuppressFinalize(this);
            this.Dispose(true);
        }

        protected void Dispose(bool disposing)
        {
            if (__disposed) return;
            if (disposing)
            {

            }

            monitoringCancellation = true;
            foreach (var observer in this.observers)
                observer.OnCompleted();

            if (this.listeningStream != null)
                this.listeningStream.Close();

            //base.Dispose();
            __disposed = true;
        }

        ~TwitterStreamMonitor()
        {
            this.Dispose(false);
        }
        #endregion

    }
}
